/**
 * File: ParallelProcessorService.java
 * Created by: mhaimel
 * Created on: 30 Mar 2009
 * CVS:  $Id: ParallelProcessorService.java,v 1.5 2009/11/24 15:50:03 mhaimel Exp $
 */
package uk.ac.ebi.velvet.exec.impl;

import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import uk.ac.ebi.curtain.utils.concurrent.UtilsCollect;
import uk.ac.ebi.velvet.exec.ProcessorService;

import com.google.inject.Singleton;

/**
 * @author mhaimel
 *
 */
@Singleton
public class ParallelProcessorService implements ProcessorService {
	private final String CATEGORY_DEFAULT = "default";
	
	private final ConcurrentMap<String, Integer> category2Threads = UtilsCollect.newConcurrentMap();
	private final ConcurrentMap<String, ExecutorService> category2Exec = UtilsCollect.newConcurrentMap(); 
	private final Queue<ExecutorService> stoppedList = new ConcurrentLinkedQueue<ExecutorService>();
	private final CopyOnWriteArraySet<Process> processSet = new CopyOnWriteArraySet<Process>();
	
	private Log log = LogFactory.getLog(this.getClass());
	
	public ParallelProcessorService() {
		registerShutdownhook();
	}


	private void registerShutdownhook() {
		Runtime.getRuntime().addShutdownHook(new Thread() {
		    public void run() { 
		    	shutdown(true);
		    	terminate();
		    }
		});
	}
	
	@Override
	public <T> Future<T> process(Callable<T> task) {
		return process(getDefaultCategory(),task);
	}

	private String getDefaultCategory() {
		return CATEGORY_DEFAULT;
	}

	@Override
	public <T> Future<T> process(String category, Callable<T> task) {
		return getService(category).submit(task);
	}

	@Override
	public void process(Runnable task) {
		process(getDefaultCategory(),task);
	}

	@Override
	public void process(String category, Runnable task) {
		getService(category).submit(task);
	}

	@Override
	public void shutdown(String category) {
		shutdown(category,false);
	}
	
	private synchronized ExecutorService getService(String category) throws IllegalArgumentException{
		ExecutorService srv = category2Exec.get(category);
		if(null == srv){
			NamedThreadFactory threadFactory = new NamedThreadFactory(category);
			if(category.equals(getDefaultCategory())){
				getLog().debug("Create new DEFAULT Thread pool for " + category + " - Cached!!!");
				srv = Executors.newCachedThreadPool(threadFactory);
			} else {		
				Integer threads = category2Threads.get(category);
				if(null == threads){
					throw new IllegalArgumentException(
							"Category " + category + " not registered!!!");
				}
				getLog().debug("Create new Thread pool for " + category + " of " + threads);
				srv = Executors.newFixedThreadPool(threads, threadFactory);
			}
			category2Exec.put(category, srv);
		}
		return srv;
	}

	@Override
	public void registerCategory(String category, Integer parallel) {
		category2Threads.put(category, parallel);
	}	
	
	private Log getLog() {
		return log;
	}

	@Override
	public void shutdown(String category, boolean force) {
		getLog().debug("Shutdown Thread pool " + category+ (force?" - FORCE":""));
		ExecutorService srv = category2Exec.remove(category);
		if(null != srv){
			shutdown(srv,force);
			stoppedList.add(srv);
		}
	}
	
    private void shutdown(ExecutorService srv, boolean force) {
    	if(null != srv && !srv.isShutdown()){
			if(force){
				srv.shutdownNow();
			} else {
				srv.shutdown();
			}
		}
	}
    
    private void terminate() {
		while(!processSet.isEmpty()){
			Process[] arr = processSet.toArray(new Process[0]);
			for(Process p : arr){
				processSet.remove(p);
				try{
					p.destroy();
				} catch (Exception e) {
					getLog().trace(e);
				}
			}
		}
    	while(!stoppedList.isEmpty()){
    		terminate(stoppedList.poll());
    	}
    }
    
    private void terminate(ExecutorService srv) {
    	if(null != srv && !srv.isShutdown() && !srv.isTerminated()){
			try {
				srv.awaitTermination(10, TimeUnit.SECONDS);
			} catch (InterruptedException e) {
				getLog().trace(e);
			}
		}
	}


	static class NamedThreadFactory implements ThreadFactory {
        static final AtomicInteger poolNumber = new AtomicInteger(1);
        final ThreadGroup group;
        final AtomicInteger threadNumber = new AtomicInteger(1);
        final String namePrefix;

        NamedThreadFactory(String name) {
            SecurityManager s = System.getSecurityManager();
            group = (s != null)? s.getThreadGroup() :
                                 Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" 
            			+ name 
            			+ "-"
            			+ poolNumber.getAndIncrement() 
            			+ "-thread-";
        }

        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);
            if (!t.isDaemon())
                t.setDaemon(true);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }

	@Override
	public synchronized void shutdown(boolean force) {
		List<String> catList = new ArrayList<String>(category2Exec.keySet());
		if(force && !catList.isEmpty()){
			getLog().info("FORCE shutdown Thread pool queues: " + catList);
		}
		for(String cat : catList){
			shutdown(cat, force);
		}
	}

	@Override
	public void add(Process process) {
		processSet.add(process);
	}


	@Override
	public void remove(Process process) {
		processSet.remove(process);
	}
		
}
